package com.lagou.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FamilyFilter;
import org.apache.hadoop.hbase.filter.ValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;

/**
 * @Author Cao Dun
 * @Date 2021/3/24 21:29
 * @Version 1.0
 * @Description
 */
public class FriendClient {

    //配置信息
    Configuration conf=null;
    //hbase连接
    Connection conn=null;
    //hbaseadmin用户
    HBaseAdmin admin=null;

    private String tableName;

    private boolean tableExists=false;

    public FriendClient(String tableName) {
        this.tableName=tableName;
    }

    /**
     * 获取hbase连接
     */
    public void initConnect() throws IOException {
        //创建配置类
        conf = HBaseConfiguration.create();
        //设置zk集群
        conf.set("hbase.zookeeper.quorum","linux121,linux122");
        //设置zk客户端端口为2181
        conf.set("hbase.zookeeper.property.clientPort","2181");
        //获取连接
        conn= ConnectionFactory.createConnection(conf);
        //获取Hbase用户
        admin= (HBaseAdmin) conn.getAdmin();
    }

    /**
     * 创建hbase表
     */
    public void createTable() throws IOException {
        if(admin.tableExists(TableName.valueOf(tableName))){
            System.out.println("friends is exists");
            this.tableExists = true;
            return;
        }

        //创建表描述器
        HTableDescriptor friends = new HTableDescriptor(TableName.valueOf(tableName));
        //设置列族描述器
        friends.addFamily(new HColumnDescriptor("info"));
        //执⾏创建操作
        admin.createTable(friends);
        System.out.println("friends 表创建成功！！");
    }

    /**
     * 向hbase表中添加数据
     */
    public void putData(){
        /**
         * 定义5个用户，假设这5个用户都互相是好友
         */
        if(this.tableExists){
            return;
        }
        ArrayList<String> arrayList = new ArrayList<String>();
        arrayList.add("uid1");
        arrayList.add("uid2");
        arrayList.add("uid3");
        arrayList.add("uid4");
        arrayList.add("uid5");

        int k = 0;

        ArrayList<Put> puts = new ArrayList<Put>();
        for (int i = 0; i < arrayList.size(); i++) {
            String rowKey = arrayList.get(i);
            System.out.print("rowKey:"+rowKey+"  ");
            k = 0;
            for (int j = 0; j < arrayList.size(); j++) {
                String colName = "friend_"+k++;
                if (i == j){
                    k--;
                    continue;
                }
                String colValue = arrayList.get(j);
                System.out.print(colName+":"+colValue+" ");
                //创建put对象
                Put put = new Put(Bytes.toBytes(rowKey));
                //添加列族，列，值
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes(colName), Bytes.toBytes(colValue));
                //把put对象添加到list中
                puts.add(put);
            }
            System.out.println();
        }


        try {
            //获取表对象
            Table friends =conn.getTable(TableName.valueOf("friends"));
            //插入数据
            friends.put(puts);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    /**
     * 获取某个列族中某个列的值为delValue对应的col
     * @param rowKey
     * @param colFamily
     * @param targetValue
     * @throws IOException
     */
    public String getValueCol(String rowKey,String colFamily,String targetValue) throws IOException {
        String targetColumn=null;

        final Table friends =conn.getTable(TableName.valueOf(tableName));
        Get get = new Get(rowKey.getBytes());
        get.addFamily(colFamily.getBytes());
        //按列值过滤
        get.setFilter(new ValueFilter(CompareFilter.CompareOp.EQUAL,new BinaryComparator(targetValue.getBytes())));

        Result result = friends.get(get);
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            System.out.println(cell);
            String cf=Bytes.toString(CellUtil.cloneFamily(cell));
            String column=Bytes.toString(CellUtil.cloneQualifier(cell));
            String value=Bytes.toString(CellUtil.cloneValue(cell));
            if (value.equals(targetValue)){
                targetColumn=column;
                continue;
            }
        }
        return targetColumn;
    }


    /**
     * 删除数据
     * @param rowKey 主键
     * @param colFamily 列族
     * @param delValue 要删除的列的值
     * @param isRecursion 是否递归关联删除,是代表要反向删除，否代表不用
     * @throws IOException
     */
    public void deleteValue(String rowKey,String colFamily,String delValue,boolean isRecursion) throws IOException {
        //需要删除的value对应的delColumn
        String delColumn = getValueCol(rowKey,colFamily,delValue);
        //需要获取⼀个table对象
        final Table friends =conn.getTable(TableName.valueOf(tableName));
        //创建删除数据delete对象
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        delete.addColumn(Bytes.toBytes(colFamily),Bytes.toBytes(delColumn));
        //删除数据
        friends.delete(delete);
        System.out.println("删除数据成功");
        //反过来删除，被删除的好友也要删除删除他的人
        if( !isRecursion){
            return;
        }
        deleteValue(delValue,colFamily,rowKey,false);
    }

    /**
     * 释放连接
     */
    @After
    public void release(){
        if(admin != null){
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        if(conn != null){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
